package org.example;

import java.io.ByteArrayInputStream;
import java.io.StringReader;
import java.sql.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

public class DataTransfer {
  private static final int THREAD_POOL_SIZE = 1; // Number of threads to handle inserts
  private static final int MAX_QUEUE_SIZE = 10000; // Maximum number of rows to buffer
  private final Queue<Map<String, Object>> cache = new ConcurrentLinkedQueue<>();
  private final AtomicInteger currentQueueSize = new AtomicInteger(0);
  private final AtomicLong rowCount = new AtomicLong();

  private List<String> columns;

  private Map<String, String> columnTypes;

  public void transferData(Connection sourceConn, Connection targetConn, String sourceTable, String targetTable,
                           String condition, int batchSize) throws SQLException {

    // 获取表的列信息，识别LONG类型字段
    columns = new ArrayList<>();
    columnTypes = new HashMap<>();

    try (Statement stmt = sourceConn.createStatement()) {
      // 查询表的列信息
      ResultSet rs = stmt.executeQuery("SELECT * FROM " + sourceTable + " WHERE 1=0");
      ResultSetMetaData rsMetaData = rs.getMetaData();
      int columnCount = rsMetaData.getColumnCount();

      for (int i = 1; i <= columnCount; i++) {
        columns.add(rsMetaData.getColumnName(i));
        columnTypes.put(rsMetaData.getColumnName(i), rsMetaData.getColumnTypeName(i));
      }
    }

    System.out.println("字段类型: " + columnTypes);

    // 添加从Java代码中赋值的额外字段
    columns.add("ODS_ROWID"); columnTypes.put("ODS_ROWID", "VARCHAR2");
    columns.add("ODS_CREATETIME"); columnTypes.put("ODS_CREATETIME", "TIMESTAMP");
    columns.add("ODS_COMMITTIME"); columnTypes.put("ODS_COMMITTIME", "TIMESTAMP");
    columns.add("ODS_UPDATETIME"); columnTypes.put("ODS_UPDATETIME", "TIMESTAMP");
    columns.add("ODS_LAG"); columnTypes.put("ODS_LAG", "VARCHAR2");
    columns.add("ODS_OPTYPE"); columnTypes.put("ODS_OPTYPE", "VARCHAR2");

    String selectSql = String.format("SELECT t.*, ROWID AS ODS_ROWID FROM %s t WHERE %s", sourceTable, condition);

    System.out.println("查询语句: " + selectSql);

    // 设置事务隔离级别为 READ_COMMITTED
    sourceConn.setTransactionIsolation(Connection.TRANSACTION_READ_COMMITTED);

    // 禁用目标数据库连接的自动提交功能
    targetConn.setAutoCommit(false);

    // 记录开始时间
    long startTime = System.currentTimeMillis();
    System.out.println("开始时间: " + new java.util.Date());

    // Create a thread pool for handling inserts
    ExecutorService executor = Executors.newFixedThreadPool(THREAD_POOL_SIZE);

    // Start a thread to handle data insertion
    Future<?> insertFuture = executor.submit(() -> {
      System.out.println("Insert task started");
      try {
        processInsert(targetConn, targetTable, batchSize);
      } catch (SQLException e) {
        e.printStackTrace();
        // Handle exception
      } catch (InterruptedException e) {
        e.printStackTrace();
        throw new RuntimeException(e);
      }
    });

    // 获取源表的列信息
    try (PreparedStatement selectStmt = sourceConn.prepareStatement(selectSql);
         ResultSet rs = selectStmt.executeQuery()) {

      ResultSetMetaData rsMetaData = rs.getMetaData();
      int columnCount = rsMetaData.getColumnCount();

      int count = 0;
      while (rs.next()) {
        Map<String, Object> row = new HashMap<>();
        for (int i = 1; i <= columnCount; i++) {
          row.put(rsMetaData.getColumnName(i), rs.getObject(i));
        }
        row.put("ODS_CREATETIME", new Timestamp(System.currentTimeMillis()));
        row.put("ODS_COMMITTIME", null);
        row.put("ODS_UPDATETIME", new Timestamp(System.currentTimeMillis()));
        row.put("ODS_LAG", "0");
        row.put("ODS_OPTYPE", "INSERT");

        // 控制队列大小，避免队列无限增长
        while (currentQueueSize.get() >= MAX_QUEUE_SIZE) {
          Thread.sleep(50); // 阻塞等待消费者处理
        }

        cache.add(row);
        currentQueueSize.incrementAndGet();
        count++;
      }

      // Signal the end of data to the insertion thread
      cache.add(new HashMap<>()); // Use empty map to signal end

      insertFuture.get(); // Wait for the insertion thread to complete

      long endTime = System.currentTimeMillis();
      long totalTimeMillis = endTime - startTime;
      double totalTimeSeconds = totalTimeMillis / 1000.0;
      System.out.printf("总耗时: %.1f 秒\n", totalTimeSeconds);
      double avgInsertPerSecond = count / totalTimeSeconds;
      System.out.printf("平均每秒插入数据: %.1f 条\n", avgInsertPerSecond);
    } catch (InterruptedException | ExecutionException e) {
      e.printStackTrace();
      // Handle exception
    } finally {
      targetConn.setAutoCommit(true);
      executor.shutdown(); // Shut down the thread pool
    }
  }

  private void processInsert(Connection targetConn, String targetTable, int batchSize) throws SQLException, InterruptedException {
    String insertSql = String.format("INSERT /*+ append nologging */ INTO %s (%s) VALUES (%s)",
        targetTable,
        String.join(", ", columns),
        String.join(", ", columns.stream().map(col -> "?").toArray(String[]::new)));

    System.out.println("插入语句: " + insertSql);

    try (PreparedStatement insertStmt = targetConn.prepareStatement(insertSql)) {
      int count = 0;

      while (true) {
        Map<String, Object> row = cache.poll();
        if (row == null) continue;

        currentQueueSize.decrementAndGet(); // 减少队列大小计数

        if (row.isEmpty()) {
          // End of data
          if (count > 0) {
            insertStmt.executeBatch();
            targetConn.commit();
            System.out.println("已插入 " + count + " 条数据");
            System.out.println("缓存记录量：" + cache.size() + "，当前时间: " + new java.util.Date());
          }
          System.out.println("End. count: " + count);
          break;
        }

        for (int i = 0; i < columns.size(); i++) {
          int index = i + 1;
          Object value = row.get(columns.get(i));
          String columnType = columnTypes.get(columns.get(i));
          if (value == null) {
            insertStmt.setNull(index, Types.NULL);
          } else {
            if ("XMLTYPE".equalsIgnoreCase(columnType)) {
              SQLXML sqlxml = targetConn.createSQLXML();
              sqlxml.setString((String) value);
              insertStmt.setSQLXML(index, sqlxml);
            } else if ("LONG".equalsIgnoreCase(columnType)) {
              if (value instanceof String) {
                insertStmt.setBytes(index, ((String) value).getBytes());
              } else {
                insertStmt.setCharacterStream(index, new StringReader((String) value));
              }
            } else if ("CLOB".equalsIgnoreCase(columnType)) {
              insertStmt.setClob(index, new StringReader((String) value));
            } else if ("BLOB".equalsIgnoreCase(columnType) && value instanceof byte[]) {
              insertStmt.setBlob(index, new ByteArrayInputStream((byte[]) value));
            } else {
              insertStmt.setObject(index, value);
            }
          }
        }

        insertStmt.addBatch();

        if (++count % batchSize == 0) {
          insertStmt.executeBatch();
          targetConn.commit();
          System.out.println("已插入 " + count + " 条数据");
          System.out.println("缓存记录量：" + cache.size() + "，当前时间: " + new java.util.Date());
        }
        //rowCount.incrementAndGet();
      }
    } catch (Exception e) {
      targetConn.rollback();
      e.printStackTrace();
      throw e;
    }
  }
}
